/*
 * Copyright (c) China Telecom Cloud Technology Co., Ltd. 2024-2025. All rights reserved.
 */
#include "ThreadPool/ThreadPool.h"


ThreadPool::ThreadWorker::ThreadWorker(ThreadPool *pool, 
            const int id) : _pool(pool), _id(id) {}


void ThreadPool::ThreadWorker::operator()()
{
    std::function<void()> func;
    bool dequeued;

    while (!_pool->_shutdown) {
        {
            std::unique_lock<std::mutex> lock(_pool->_mutex);
            // 如果队列里没有任务，则进入休眠，减少资源消耗
            // wait会释放锁
            // TODO(yangxianpku: 2025.08.18): 使用wait+lambda避免虚假唤醒
            _pool->_has_task.wait(lock, 
                                [this]() { 
                                    return _pool->_shutdown    ||  
                                        !_pool->_queue.empty();
                                }
                            );

            // 如果已经关闭且队列为空，立即退出
            if (_pool->_shutdown && _pool->_queue.empty()) {
                break;
            }
            
            dequeued = _pool->_queue.dequeue(func);
        }

        // 取出来一份任务则将其运行
        if (dequeued) {
            func();
        }
    }
}





ThreadPool::ThreadPool(const int n_threads) 
    : _threads(std::vector<std::thread>(n_threads)), 
      _shutdown(false) {}




ThreadPool::~ThreadPool() {
    shutdown();
}




void ThreadPool::init(){
    for (int i = 0; i < _threads.size(); ++i) {
        _threads[i] = std::thread(ThreadWorker(this, i));
    }
}





void ThreadPool::shutdown() {
    if (_shutdown) return;
    
    // TODO(yangxianpku: 2025.08.21):  考虑_shutdown的安全， 使用shared_mutex
    _shutdown = true;
    _has_task.notify_all();
    
    for (auto &thread : _threads) {
        if (thread.joinable()) {
            thread.join();
        }
    }
}